package com.geekstudy.week13.kafka;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 *
 *
 *
 * https://www.jianshu.com/p/6ce5d9a96113?utm_campaign=hugo
 *
 *
 *
 */
@Service
public class KafkaProducerService {

    final Logger logger = LoggerFactory.getLogger(KafkaProducerService.class);

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public static final String TopicDemo1 = "topic.demo1";
    public static final String TopicGroup1 = "topic.group1";

    public void send(String message, String topic) {
        logger.info(" send:{} ", message);

        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, message);
        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {

            @Override
            public void onSuccess(SendResult<String, String> stringSendResult) {
                logger.info("onSuccess sendResult:{} ", stringSendResult);
            }

            @Override
            public void onFailure(Throwable throwable) {
                logger.info("onFailur throwable: {}", throwable);

            }
        });


    }


}
